package org.apache.flink.runtime.rest.messages;

import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.ConsumingStatus;
import org.apache.flink.util.ERException;
import org.apache.flink.util.Preconditions;

/**
 * Exception indicating that the Flink job with the given job ID has terminated without
 * cancellation.
 */
public class FlinkJobTerminatedWithoutCancellationException extends ERException {

    private static final long serialVersionUID = 2294698055059659025L;

    private final ConsumingStatus jobStatus;

    public FlinkJobTerminatedWithoutCancellationException(ConsumerID jobId, ConsumingStatus jobStatus) {
        super(
                String.format(
                        "Flink job (%s) was not canceled, but instead %s.",
                        jobId, assertNotCanceled(jobStatus)));
        this.jobStatus = jobStatus;
    }

    public ConsumingStatus getJobStatus() {
        return jobStatus;
    }

    private static ConsumingStatus assertNotCanceled(ConsumingStatus jobStatus) {
        Preconditions.checkState(jobStatus != ConsumingStatus.CANCELED);
        return jobStatus;
    }
}
